package com.poetic.login.fail.detect;

import com.poetic.login.fail.detect.domain.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * <pre>
 *  状态编程 实现：
 *      要按照用户 ID 分流，然后遇到登录失败的事件时将其保存在 ListState 中，
 * 然后设置一个定时器，2 秒后触发。定时器触发时检查状态中的登录失败事件个数，
 * 如果大于等于 2，那么就输出报警信息。
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class LoginFailDetectByState {
    public static void main(String[] args) throws Exception {
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/LoginLog.csv";
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        // 5402,83.149.11.115,success,1558430815
        env.readTextFile(path)
                .map(new MapFunction<String, LoginEvent>() {
                    @Override
                    public LoginEvent map(String value) throws Exception {
                        String[] line = value.split("\\,");
                        LoginEvent event = LoginEvent.builder()
                                .userId(Long.valueOf(line[0]))
                                .ip(line[1])
                                .eventType(line[2])
                                .eventTime(Long.valueOf(line[3]))
                                .build();

                        return event;
                    }
                })
                .filter(new FilterFunction<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent event) throws Exception {
                        return event.getEventType().equals("fail");
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                    @Override
                    public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                        return element.getEventTime();
                    }
                }))
                .keyBy(new KeySelector<LoginEvent, Long>() {
                    @Override
                    public Long getKey(LoginEvent value) throws Exception {
                        return value.getUserId();
                    }
                })
                .process(new KeyedProcessFunction<Long, LoginEvent, LoginEvent>() {

                    private ListState<LoginEvent> loginState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ListStateDescriptor<LoginEvent> loginStateDesc = new ListStateDescriptor<LoginEvent>("loginState",
                                TypeInformation.of(new TypeHint<LoginEvent>() {
                                }));

                        loginState = getRuntimeContext().getListState(loginStateDesc);
                    }

                    @Override
                    public void processElement(LoginEvent value, Context ctx, Collector<LoginEvent> out) throws Exception {
                        loginState.add(value);
                        // 定义定时器两秒后触发，统计错误条数
                        ctx.timerService().registerEventTimeTimer(value.getEventTime() * 1000 + 2 * 1000);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginEvent> out) throws Exception {

                        List<LoginEvent> loginEventList = new ArrayList<>();
                        Iterable<LoginEvent> loginEvents = loginState.get();
                        loginEvents.forEach(login -> loginEventList.add(login));

                        loginState.clear();

                        if (loginEventList.size() > 1) {
                            out.collect(loginEventList.get(0));
                        }
                    }
                })
                .print();

        env.execute("login fail ");
    }
}
